package com.ydl.learning.flink.demo.time

import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks
import org.apache.flink.streaming.api.watermark.Watermark
import org.slf4j.{Logger, LoggerFactory}

/**
 * 自定义水印
 *
 * @author ydl
 * @since 2020/11/23
 */
class MyWaterMark extends AssignerWithPeriodicWatermarks[String] {
  private val log: Logger = LoggerFactory.getLogger(classOf[MyWaterMark])
  /**
   * 延迟时间
   */
  private val maxOutOrderness = 5000L

  /**
   * flink过一段时间便会调一次该函数获取水印放入数据流中
   * 时间间隔由streamEnv.getConfig.setAutoWatermarkInterval设置，默认100毫秒
   */
  override def getCurrentWatermark: Watermark = {
    val sysMilssecons = System.currentTimeMillis()
    //log.info("获取水印:{}", sysMilssecons)
    //设置水印延迟5秒
    new Watermark(sysMilssecons - maxOutOrderness)

  }

  /**
   * 每条记录都会调用 来获得even time 在生产的数据中 even_time放在字符串的第一个字段 用空格分割
   */
  override def extractTimestamp(element: String, previousElementTimestamp: Long): Long = {
    log.info("设置时间戳:{},原有时间戳:{}{}", element, previousElementTimestamp.toString, "")
    element.split(" ").head.toLong
  }
}
